# pip install kafka-python==2.0.2

from kafka import KafkaProducer, KafkaConsumer
import json
import traceback
from app.conf import settings
from app.api.generate.utils.generate_queue.process_message import msg_process


ip = settings.KAFKA_HOST
port = settings.KAFKA_PORT
topic_list = settings.KAFKA_TOPIC_LIST
group = settings.KAFKA_GROUP
host = ip + ':' + port

# -------------------------------------------------------------------------------------------------------------------------------------------------------------------------

producer = KafkaProducer(bootstrap_servers=host, api_version=(0, 10, 2), max_request_size=524288000)  # 连接kafka


def kafka_send_msg(send_topic, data):
    print("send message")
    producer.send(send_topic, json.dumps(data).encode('utf-8')).get(timeout=90)


# -------------------------------------------------------------------------------------------------------------------------------------------------------------------------

def start_listen_kafka():
    consumer = KafkaConsumer(
        bootstrap_servers=[host], 
        group_id=group, 
        value_deserializer=lambda m: json.loads(m.decode('utf-8')),
        # max_poll_interval_ms=3000000,
        # session_timeout_ms=30000,
        )
    print("kafka消费者连接成功")
    # print(consumer)
    consumer.subscribe(topics=topic_list)
    print("client IP", host,  "订阅消息：", topic_list)
    for msg in consumer:
        v = msg.value
        accept_topic = msg.topic
        try:
            msg_process(accept_topic, v)
        except Exception as e:
            print(v)
            print(traceback.format_exc())
            




